package org.jgroups.stack;

import org.jgroups.Address;
import org.jgroups.Message;
import org.jgroups.protocols.FragHeader3;
import org.jgroups.util.TimeScheduler;
import org.jgroups.util.Util;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;

/**
 * Created by IntelliJ IDEA.
 * User: thsieh
 * Date: Mar 16, 2009
 * Time: 2:48:44 PM
 * To change this template use File | Settings | File Templates.
 */
public class FragRetransmitter {
    private static final long SEC=39;
    /** Default retransmit intervals (ms) - exponential approx. */
    private Interval RETRANSMIT_TIMEOUTS=new StaticInterval(SEC, 7*SEC, 18*SEC);
    //private Address sender=null;
    private final ConcurrentMap<Long,Task> msgs=new ConcurrentHashMap<Long,Task>(11);
    private RetransmitCommand cmd=null;
    private TimeScheduler timer=null;
    protected static final Log log= LogFactory.getLog(FragRetransmitter.class);


    /** Retransmit command (see Gamma et al.) used to retrieve missing messages */
    public interface RetransmitCommand {
        /**
         * Get the missing messages between frag_id
         * retransmit message to destination <code>sender</code> (nak-based scheme), or by
         * retransmitting the missing message(s) to <code>sender</code> (ack-based scheme).
         * @param messageKey
         *               (nak-based scheme), or to which the message will be retransmitted (ack-based scheme).
         */
        void retransmit(long messageKey, Message msg);
    }


    /**
     * Create a new Retransmitter associated with the given sender address
     //* @param sender the address from which retransmissions are expected or to which retransmissions are sent
     * @param cmd the retransmission callback reference
     * @param sched retransmissions scheduler
     */
    public FragRetransmitter(RetransmitCommand cmd, TimeScheduler sched) {
        this.cmd=cmd;
        timer=sched;
        //init(sender, cmd, sched);
    }



    public void setRetransmitTimeouts(Interval interval) {
        if(interval != null)
            RETRANSMIT_TIMEOUTS=interval;
    }


    /**
     * Add the given range [first_seqno, last_seqno] in the list of
     * entries eligible for retransmission. If first_seqno > last_seqno,
     * then the range [last_seqno, first_seqno] is added instead
     * <p>
     * If retransmitter thread is suspended, wake it up
     */
    public void add(long key, Message msg) {

        Task new_task=new Task(key, RETRANSMIT_TIMEOUTS.copy(), cmd, msg);
        Task old_task=msgs.putIfAbsent(key, new_task);
        if(old_task == null) // only schedule if we actually *added* the new task !
            new_task.doSchedule(timer); // Entry adds itself to the timer
        else if ( log.isInfoEnabled() ) log.info("Key is in queue " + key );
    }

    /**
     * Remove the given sequence number from the list of seqnos eligible
     * for retransmission. If there are no more seqno intervals in the
     * respective entry, cancel the entry from the retransmission
     * scheduler and remove it from the pending entries
     */
    public int remove(long key) {
        Task task=msgs.remove(key);
        if(task != null) {
            task.cancel();
            return task.getNumRetransmits();
        }
        return -1;
    }

    /**
     * Reset the retransmitter: clear all msgs and cancel all the
     * respective tasks
     */
    public void reset() {
        for(Task task: msgs.values())
            task.cancel();
        msgs.clear();
    }


    public void stop() {
        reset();
    }


    public String toString() {
        int size=size();
        StringBuilder sb=new StringBuilder();
        sb.append(size).append(" messages to retransmit: ").append(msgs.keySet());
        return sb.toString();
    }


    public int size() {
        return msgs.size();
    }




    /* ------------------------------- Private Methods -------------------------------------- */

    /**
     * Init this object
     *
     * @param sender the address from which retransmissions are expected
     * @param cmd the retransmission callback reference
     * @param sched retransmissions scheduler
     */
//    private void init(Address sender, RetransmitCommand cmd, TimeScheduler sched) {
//        this.sender=sender;
//        this.cmd=cmd;
//        timer=sched;
//    }


    /* ---------------------------- End of Private Methods ------------------------------------ */



    /**
     * The retransmit task executed by the scheduler in regular intervals
     */
    private static class Task implements TimeScheduler.Task {
        private final Interval    intervals;
        private long              key=-1;
        private Future future;
        //private Address           sender=null;
        protected int             num_retransmits=0;
        private RetransmitCommand command;
        //private FragHeader3 header;
        private Message msg;

        protected Task(long key, Interval intervals, RetransmitCommand cmd, Message msg) {
            this.key=key;
            this.intervals=intervals;
            this.command=cmd;
            this.msg= msg;
        }

        public int getNumRetransmits() {
            return num_retransmits;
        }

        public long nextInterval() {
            return intervals.next();
        }

        public void doSchedule(TimeScheduler timer) {
            future=timer.scheduleWithDynamicInterval(this);
        }

        public int getSize() {
            return ((StaticInterval) intervals).getSize() ;
        }
        public void cancel() {
            if(future != null)
                future.cancel(false);
        }

        public void run() {
            if( num_retransmits > getSize()  ) {
                log.info("Run away thread num_retransmits " + num_retransmits+ " key " + key +" "+ this.toString() );
                cancel() ;
            }
            command.retransmit( key, msg);
            num_retransmits++;
            // make sure it did not run more than twice
        }

        public String toString() {
            return String.valueOf(key);
        }
    }






    static class MyXmitter implements Retransmitter.RetransmitCommand {

        public void retransmit(long first_seqno, long last_seqno, Address sender) {
            System.out.println("-- " + new java.util.Date() + ": retransmit(" + first_seqno + ", " +
                               last_seqno + ", " + sender + ')');
        }
    }

    static void sleep(long timeout) {
        Util.sleep(timeout);
    }

}
